﻿// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.

using System.Buffers;
using System.Diagnostics;
using System.Runtime.CompilerServices;

namespace System.IO.Pipelines
{
    public sealed partial class Pipe
    {
        // Mutable struct! Don't make this readonly
        private BufferSegmentStack _bufferSegmentPool;

        // The options instance
        private readonly PipeOptions _options;
        private readonly object _sync = new object();

        private int MinimumSegmentSize => _options.MinimumSegmentSize;
        private long PauseWriterThreshold => _options.PauseWriterThreshold;
        private long ResumeWriterThreshold => _options.ResumeWriterThreshold;

        private object SyncObj => _sync;

        // The number of bytes flushed but not consumed by the reader
        private long _unconsumedBytes;

        // The number of bytes written but not flushed
        private long _unflushedBytes;

        private long _lastExaminedIndex = -1;

        // The read head which is the start of the PipeReader's consumed bytes
        private BufferSegment _readHead;
        private int _readHeadIndex;

        private bool _disposed;

        // The extent of the bytes available to the PipeReader to consume
        private BufferSegment _readTail;
        private int _readTailIndex;
        private int _minimumReadBytes;

        // The write head which is the extent of the PipeWriter's written bytes
        private BufferSegment _writingHead;
        private Memory<byte> _writingHeadMemory;
        private int _writingHeadBytesBuffered;

       
        internal long Length => _unconsumedBytes;

        private bool IsCompleted { get { return _disposed; } }

        /// <summary>Initializes a new instance of the <see cref="System.IO.Pipelines.Pipe" /> class with the specified options.</summary>
        /// <param name="options">The set of options for this pipe.</param>
        public Pipe(PipeOptions options)
        {
            if (options == null)
            {
                ThrowHelper.ThrowArgumentNullException(ExceptionArgument.options);
            }

            _bufferSegmentPool = new BufferSegmentStack(options.InitialSegmentPoolSize);

            _options = options;
        }

        private void ResetState()
        {
            _readTailIndex = 0;
            _readHeadIndex = 0;
            _lastExaminedIndex = -1;
            _unflushedBytes = 0;
            _unconsumedBytes = 0;
        }

        internal Memory<byte> GetMemory(int sizeHint)
        {
            if (IsCompleted)
            {
                ThrowHelper.ThrowInvalidOperationException_NoWritingAllowed();
            }

            if (sizeHint < 0)
            {
                ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.minimumSize);
            }

            AllocateWriteHeadIfNeeded(sizeHint);

            return _writingHeadMemory;
        }

        internal Span<byte> GetSpan(int sizeHint)
        {
            if (IsCompleted)
            {
                ThrowHelper.ThrowInvalidOperationException_NoWritingAllowed();
            }

            if (sizeHint < 0)
            {
                ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.minimumSize);
            }

            AllocateWriteHeadIfNeeded(sizeHint);

            return _writingHeadMemory.Span;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        private void AllocateWriteHeadIfNeeded(int sizeHint)
        {
            // If writing is currently active and enough space, don't need to take the lock to just set WritingActive.
            // IsWritingActive is needed to prevent the reader releasing the writers memory when it fully consumes currently written.
            if (!IsCompleted || _writingHeadMemory.Length == 0 || _writingHeadMemory.Length < sizeHint)
            {
                AllocateWriteHeadSynchronized(sizeHint);
            }
        }

        private void AllocateWriteHeadSynchronized(int sizeHint)
        {
            lock (SyncObj)
            {
                if (_writingHead == null)
                {
                    // We need to allocate memory to write since nobody has written before
                    BufferSegment newSegment = AllocateSegment(sizeHint);

                    // Set all the pointers
                    _writingHead = _readHead = _readTail = newSegment;
                    _lastExaminedIndex = 0;
                }
                else
                {
                    int bytesLeftInBuffer = _writingHeadMemory.Length;

                    if (bytesLeftInBuffer == 0 || bytesLeftInBuffer < sizeHint)
                    {
                        if (_writingHeadBytesBuffered > 0)
                        {
                            // Flush buffered data to the segment
                            _writingHead.End += _writingHeadBytesBuffered;
                            _writingHeadBytesBuffered = 0;
                        }

                        BufferSegment newSegment = AllocateSegment(sizeHint);

                        _writingHead.SetNext(newSegment);
                        _writingHead = newSegment;
                    }
                }
            }
        }

        private BufferSegment AllocateSegment(int sizeHint)
        {
            Debug.Assert(sizeHint >= 0);
            BufferSegment newSegment = CreateSegmentUnsynchronized();

            MemoryPool<byte> pool = null;
            int maxSize = -1;

            if (!_options.IsDefaultSharedMemoryPool)
            {
                pool = _options.Pool;
                maxSize = pool.MaxBufferSize;
            }

            if (sizeHint <= maxSize)
            {
                // Use the specified pool as it fits. Specified pool is not null as maxSize == -1 if _pool is null.
                newSegment.SetOwnedMemory(pool!.Rent(GetSegmentSize(sizeHint, maxSize)));
            }
            else
            {
                // Use the array pool
                int sizeToRequest = GetSegmentSize(sizeHint);
                newSegment.SetOwnedMemory(ArrayPool<byte>.Shared.Rent(sizeToRequest));
            }

            _writingHeadMemory = newSegment.AvailableMemory;

            return newSegment;
        }

        private int GetSegmentSize(int sizeHint, int maxBufferSize = int.MaxValue)
        {
            // First we need to handle case where hint is smaller than minimum segment size
            sizeHint = Math.Max(MinimumSegmentSize, sizeHint);
            // After that adjust it to fit into pools max buffer size
            int adjustedToMaximumSize = Math.Min(maxBufferSize, sizeHint);
            return adjustedToMaximumSize;
        }

        private BufferSegment CreateSegmentUnsynchronized()
        {
            if (_bufferSegmentPool.TryPop(out BufferSegment segment))
            {
                return segment;
            }

            return new BufferSegment();
        }

        private void ReturnSegmentUnsynchronized(BufferSegment segment)
        {
            Debug.Assert(segment != _readHead, "Returning _readHead segment that's in use!");
            Debug.Assert(segment != _readTail, "Returning _readTail segment that's in use!");
            Debug.Assert(segment != _writingHead, "Returning _writingHead segment that's in use!");

            if (_bufferSegmentPool.Count < _options.MaxSegmentPoolSize)
            {
                _bufferSegmentPool.Push(segment);
            }
        }

        internal bool CommitUnsynchronized()
        {
            if (_unflushedBytes == 0)
            {
                // Nothing written to commit
                return false;
            }

            // Update the writing head
            Debug.Assert(_writingHead != null);
            _writingHead.End += _writingHeadBytesBuffered;

            // Always move the read tail to the write head
            _readTail = _writingHead;
            _readTailIndex = _writingHead.End;

            long oldLength = _unconsumedBytes;
            _unconsumedBytes += _unflushedBytes;

            bool resumeReader = true;

            if (_unconsumedBytes < _minimumReadBytes)
            {
                // Don't yield the reader if we haven't written enough
                resumeReader = false;
            }
            // We only apply back pressure if the reader isn't paused. This is important
            // because if it is blocked then this could cause a deadlock (if resumeReader is false).
            // If we are resuming the reader, then we can look at the pause threshold to know
            // if we should pause the writer.
            else if (PauseWriterThreshold > 0 &&
                oldLength < PauseWriterThreshold &&
                _unconsumedBytes >= PauseWriterThreshold &&
                !IsCompleted)
            {
                //_writerAwaitable.SetUncompleted();
            }

            _unflushedBytes = 0;
            _writingHeadBytesBuffered = 0;

            return resumeReader;
        }

        public void Advance(int bytes)
        {
            lock (SyncObj)
            {
                if ((uint)bytes > (uint)_writingHeadMemory.Length)
                {
                    ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.bytes);
                }

                // If the reader is completed we no-op Advance but leave GetMemory and FlushAsync alone
                if (IsCompleted)
                {
                    return;
                }

                AdvanceCore(bytes);
            }
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        private void AdvanceCore(int bytesWritten)
        {
            _unflushedBytes += bytesWritten;
            _writingHeadBytesBuffered += bytesWritten;
            _writingHeadMemory = _writingHeadMemory.Slice(bytesWritten);
        }

        internal ValueTask FlushAsync(CancellationToken cancellationToken)
        {
            lock (SyncObj)
            {
                PrepareFlush();
            }

            return new ValueTask();
        }

        private void PrepareFlush()
        {
            var completeReader = CommitUnsynchronized();
        }

        internal void CompleteWriter(Exception exception)
        {
            bool readerCompleted;

            lock (SyncObj)
            {
                // Commit any pending buffers
                CommitUnsynchronized();

                readerCompleted = this.IsCompleted;
            }

            if (readerCompleted)
            {
                CompletePipe();
            }
        }

        public void AdvanceReader(in SequencePosition consumed)
        {
            AdvanceReader(consumed, consumed);
        }

        internal void AdvanceReader(in SequencePosition consumed, in SequencePosition examined)
        {
            // If the reader is completed
            if (IsCompleted)
            {
                ThrowHelper.ThrowInvalidOperationException_NoReadingAllowed();
            }

            // TODO: Use new SequenceMarshal.TryGetReadOnlySequenceSegment to get the correct data
            // directly casting only works because the type value in ReadOnlySequenceSegment is 0
            AdvanceReader((BufferSegment)consumed.GetObject(), consumed.GetInteger(), (BufferSegment)examined.GetObject(), examined.GetInteger());
        }

        private void AdvanceReader(BufferSegment consumedSegment, int consumedIndex, BufferSegment examinedSegment, int examinedIndex)
        {
            // Throw if examined < consumed
            if (consumedSegment != null && examinedSegment != null && BufferSegment.GetLength(consumedSegment, consumedIndex, examinedSegment, examinedIndex) < 0)
            {
                ThrowHelper.ThrowInvalidOperationException_InvalidExaminedOrConsumedPosition();
            }

            BufferSegment returnStart = null;
            BufferSegment returnEnd = null;

            //CompletionData completionData = default;

            lock (SyncObj)
            {
                var examinedEverything = false;
                if (examinedSegment == _readTail)
                {
                    examinedEverything = examinedIndex == _readTailIndex;
                }

                if (examinedSegment != null && _lastExaminedIndex >= 0)
                {
                    long examinedBytes = BufferSegment.GetLength(_lastExaminedIndex, examinedSegment, examinedIndex);
                    long oldLength = _unconsumedBytes;

                    if (examinedBytes < 0)
                    {
                        ThrowHelper.ThrowInvalidOperationException_InvalidExaminedPosition();
                    }

                    _unconsumedBytes -= examinedBytes;

                    // Store the absolute position
                    _lastExaminedIndex = examinedSegment.RunningIndex + examinedIndex;

                    Debug.Assert(_unconsumedBytes >= 0, "Length has gone negative");

                    //if (oldLength >= ResumeWriterThreshold &&
                    //    _unconsumedBytes < ResumeWriterThreshold)
                    //{
                    //    //_writerAwaitable.Complete(out completionData);
                    //}
                }

                if (consumedSegment != null)
                {
                    if (_readHead == null)
                    {
                        ThrowHelper.ThrowInvalidOperationException_AdvanceToInvalidCursor();
                        return;
                    }

                    returnStart = _readHead;
                    returnEnd = consumedSegment;

                    void MoveReturnEndToNextBlock()
                    {
                        BufferSegment nextBlock = returnEnd!.NextSegment;
                        if (_readTail == returnEnd)
                        {
                            _readTail = nextBlock;
                            _readTailIndex = 0;
                        }

                        _readHead = nextBlock;
                        _readHeadIndex = 0;

                        returnEnd = nextBlock;
                    }

                    if (consumedIndex == returnEnd.Length)
                    {
                        // If the writing head isn't block we're about to return, then we can move to the next one
                        // and return this block safely
                        if (_writingHead != returnEnd)
                        {
                            MoveReturnEndToNextBlock();
                        }
                        // If the writing head is the same as the block to be returned, then we need to make sure
                        // there's no pending write and that there's no buffered data for the writing head
                        else if (_writingHeadBytesBuffered == 0 && !IsCompleted)
                        {
                            // Reset the writing head to null if it's the return block and we've consumed everything
                            _writingHead = null;
                            _writingHeadMemory = default;

                            MoveReturnEndToNextBlock();
                        }
                        else
                        {
                            _readHead = consumedSegment;
                            _readHeadIndex = consumedIndex;
                        }
                    }
                    else
                    {
                        _readHead = consumedSegment;
                        _readHeadIndex = consumedIndex;
                    }
                }

                while (returnStart != null && returnStart != returnEnd)
                {
                    BufferSegment next = returnStart.NextSegment;
                    returnStart.ResetMemory();
                    ReturnSegmentUnsynchronized(returnStart);
                    returnStart = next;
                }
            }
        }

        internal void CompleteReader(Exception exception)
        {
            bool writerCompleted;

            lock (SyncObj)
            {
                writerCompleted = IsCompleted;
            }

            if (writerCompleted)
            {
                CompletePipe();
            }
        }

        public ReadOnlySequence<byte> Read()
        {
            if (IsCompleted)
            {
                ThrowHelper.ThrowInvalidOperationException_NoReadingAllowed();
            }

            ReadOnlySequence<byte> result;
            lock (SyncObj)
            {
                //_readerAwaitable.BeginOperation(token, s_signalReaderAwaitable, this);

                // If the awaitable is already complete then return the value result directly
                if (!IsCompleted)
                {
                    GetReadResult(out result);
                }
                else
                {
                    // Otherwise it's async
                    result = default;
                }
            }

            return result;
        }

        private void CompletePipe()
        {
            lock (SyncObj)
            {
                if (_disposed)
                {
                    return;
                }

                _disposed = true;
                // Return all segments
                // if _readHead is null we need to try return _commitHead
                // because there might be a block allocated for writing
                BufferSegment segment = _readHead ?? _readTail;
                while (segment != null)
                {
                    BufferSegment returnSegment = segment;
                    segment = segment.NextSegment;

                    returnSegment.ResetMemory();
                }

                _writingHead = null;
                _writingHeadMemory = default;
                _readHead = null;
                _readTail = null;
                _lastExaminedIndex = -1;
            }
        }

        private void GetReadResult(out ReadOnlySequence<byte> result)
        {
            // No need to read end if there is no head
            BufferSegment head = _readHead;
            if (head != null)
            {
                Debug.Assert(_readTail != null);
                // Reading commit head shared with writer
                result = new ReadOnlySequence<byte>(head, _readHeadIndex, _readTail, _readTailIndex);
            }
            else
            {
                result = default;
            }

            // Reset the minimum read bytes when read yields
            _minimumReadBytes = 0;
        }

        public void Write(ReadOnlyMemory<byte> source)
        {
            if (IsCompleted)
            {
                ThrowHelper.ThrowInvalidOperationException_NoWritingAllowed();
            }

            lock (SyncObj)
            {
                // Allocate whatever the pool gives us so we can write, this also marks the
                // state as writing
                AllocateWriteHeadIfNeeded(0);

                if (source.Length <= _writingHeadMemory.Length)
                {
                    source.CopyTo(_writingHeadMemory);

                    AdvanceCore(source.Length);
                }
                else
                {
                    // This is the multi segment copy
                    WriteMultiSegment(source.Span);
                }
                PrepareFlush();
            }
        }

        private void WriteMultiSegment(ReadOnlySpan<byte> source)
        {
            Debug.Assert(_writingHead != null);
            Span<byte> destination = _writingHeadMemory.Span;

            while (true)
            {
                int writable = Math.Min(destination.Length, source.Length);
                source.Slice(0, writable).CopyTo(destination);
                source = source.Slice(writable);
                AdvanceCore(writable);

                if (source.Length == 0)
                {
                    break;
                }

                // We filled the segment
                _writingHead.End += _writingHeadBytesBuffered;
                _writingHeadBytesBuffered = 0;

                // This is optimized to use pooled memory. That's why we pass 0 instead of
                // source.Length
                BufferSegment newSegment = AllocateSegment(0);

                _writingHead.SetNext(newSegment);
                _writingHead = newSegment;

                destination = _writingHeadMemory.Span;
            }
        }

        /// <summary>Resets the pipe.</summary>
        public void Reset()
        {
            lock (SyncObj)
            {
                if (!_disposed)
                {
                    ThrowHelper.ThrowInvalidOperationException_ResetIncompleteReaderWriter();
                }

                _disposed = false;
                ResetState();
            }
        }
    }
}
